package com.bingxu.flink.action;

import com.bingxu.flink.bean.UserBehavior;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.io.Serializable;

/**
 * @author :BingXu
 * @description :TODO
 * @date :2021/8/13 8:24
 * @modifier :
 */

public class PV_Count4 {
    public static void main(String[] args) throws Exception {
        // 运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // load data and transform Java Bean object
        DataStreamSource<String> sourceDS = env.readTextFile("D:\\Source\\JavaProject\\flinkdemo\\input\\UserBehavior.csv");
        SingleOutputStreamOperator<UserBehavior> userBehaviorDS = sourceDS.map(new MapFunction<String, UserBehavior>() {
            @Override
            public UserBehavior map(String s) throws Exception {
                String[] fields = s.split(",");
                return new UserBehavior(Long.parseLong(fields[0]),
                        Long.parseLong(fields[1]),
                        Integer.parseInt(fields[2]),
                        fields[3],
                        Long.parseLong(fields[4])
                );
            }
        });
        // 通过累加器进行处理
        SingleOutputStreamOperator<UserBehavior> filterDS = userBehaviorDS.filter(ele -> "pv".equals(ele.getBehavior()));
        SingleOutputStreamOperator<Integer> processDS = filterDS.process(new ProcessFunction<UserBehavior, Integer>() {
            private IntCounter pvCount = new IntCounter();

            @Override
            public void open(Configuration parameters) throws Exception {
                // register accumulator
                getRuntimeContext().addAccumulator("pv_count", pvCount);
            }

            @Override
            public void processElement(UserBehavior value, Context ctx, Collector<Integer> out) throws Exception {
                Accumulator<Object, Serializable> pv_count = getRuntimeContext().getAccumulator("pv_count");
                pv_count.add(1);
            }

        });

        // 获取执行的结果，累加器的结果在执行结果之中

        // 执行
        JobExecutionResult execute = env.execute();
        Object pv_count = execute.getAccumulatorResult("pv_count");
        System.out.println("pv_count = " + pv_count);
    }
}
